/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.core.impl;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.mina.common.IoSession;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.base.AbstractConnectorSupport;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.SessionDispatcher;
import com.cqcis.uip.base.framework.core.SocketClientTaskContextAfterProcessor;
import com.cqcis.uip.base.framework.core.TaskContextSessionStrategy;
import com.cqcis.uip.base.framework.handlers.ClientService;
import com.cqcis.uip.base.framework.handlers.common.AbstractClientHandler;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;

/**
 * @author huangmy
 * @date   2008-8-27
 */
public class SocketClientTaskContextAdaptor extends AbstractClientTaskContextAdaptor<Object> 
        implements ClientService, SessionDispatcher, TaskContextSessionStrategy<Object> {
	private final static String COUNTER = "COUNTER_" + SocketClientTaskContextAdaptor.class.getName();
	private Set<Task> requestSessionTasks = Collections.synchronizedSet(new HashSet<Task>()); // 请求session的task
    private ConcurrentTaskQueue<Object> requestSessions = new ConcurrentTaskQueue<Object>();
	private AbstractConnectorSupport currentRS;
	@Override
	protected AbstractTaskContext<Object> createTaskContext(
			RegisteredServiceContext serviceContext, Task task) {	
		AbstractTaskContext<Object> tc = new DefaultSocketClientTaskContext(serviceContext, task);
		tc.setRequestSessions(requestSessions);
		tc.setRequestSessionTasks(requestSessionTasks);
		tc.setConnectBeforeWrite(true); // 设置如果是短连接，则为写前建立连接
		tc.setTaskContextSessionStrategy(this);
		return tc;
	}
	
	public boolean isToDispatch(IoSession session) {	
		if (this.requestSessionTasks.size() > 0) {
			this.requestSessions.offer(session);
			return false;
		}
		return true;
	}
	
	public void offerSession(Object session) {
		if (requestSessionTasks.size() > 0) {
			requestSessions.offer(session);
		}
	}
	
	public void newSession() {
		try {
			this.currentRS.clientNew();
		} catch (Exception e) {
			throw new UIPServiceException(e);
		}
	}
	
	@Override
	protected void notifyCommitEnd(Task task) {
	}
	
	@Override
	public void start(RegisteredServiceContext serviceContext) throws Exception {		
		super.start(serviceContext);
		this.currentRS = ((AbstractConnectorSupport)serviceContext.getRegisteredService());    	
    	AbstractClientHandler handler = this.currentRS.getHandler();  
		handler.registerSessionDispatcher(this);
		this.currentRS.registerTaskContextSessionStrategy(this);
	}

	public void doMessageReceived(Object message, Task task) throws Exception {
		AtomicLong counter = (AtomicLong)task.getAttribute(COUNTER);
		if (counter == null) {
			counter = new AtomicLong();
			task.setAttribute(COUNTER, counter);
		}
		counter.incrementAndGet();
		
		super.messageSyncReceived(message, task);
	}

	public Object doMessageSend(Task task) throws Exception {
		return super.messageSyncSend(task);
	}

	/**
	 * 正常结束，通过该种方式支持在ClientBlockingService中没有进行任何操作就结束的情况
	 * 
	 */
	@Override
	protected void afterNormalFinished(Task task) {
		IoSession session = task.getSession();
		AtomicLong counter = (AtomicLong)task.removeAttribute(COUNTER);
		SocketClientTaskContextAfterProcessor afterProcessor = (SocketClientTaskContextAfterProcessor)session.getAttribute(FrameworkConstants.SOCKET_CLIENT_TASK_AFTERPROCESSOR);
		if (afterProcessor != null && (counter == null || (counter.get() <= 0))) {
			afterProcessor.afterNormalFinished(task, session);
		}
		super.afterNormalFinished(task); 
	}
}
